/**
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
 * agreements.  See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership.  The ASF licenses this file to you under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with the License.  You may obtain
 * a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */


package org.apache.zookeeper.server.quorum;

import org.apache.zookeeper.common.Time;
import org.apache.zookeeper.jmx.MBeanRegistry;
import org.apache.zookeeper.server.ZooKeeperThread;
import org.apache.zookeeper.server.quorum.QuorumCnxManager.Message;
import org.apache.zookeeper.server.quorum.QuorumPeer.LearnerType;
import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer;
import org.apache.zookeeper.server.quorum.QuorumPeer.ServerState;
import org.apache.zookeeper.server.util.ZxidUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;


/**
 * Implementation of leader election using TCP. It uses an object of the class
 * QuorumCnxManager to manage connections. Otherwise, the algorithm is push-based
 * as with the other UDP implementations.
 *
 * There are a few parameters that can be tuned to change its behavior. First,
 * finalizeWait determines the amount of time to wait until deciding upon a leader.
 * This is part of the leader election algorithm.
 *
 * 当zk集群中的一台server出现以下两种情况之一时，就会开始进入Leader选举。
 *  1、服务器初始化启动
 *  2、服务器运行期间无法和Leader保持连接。（连接超时）
 * 而当一台机器进入Leader选举流程时，当前集群也可能处于以下两种状态：
 *  1、集群中本来就已经存在一个Leader  ： 比如集群中某台机器server1启动比较晚，在它启动之前集群已经正常工作（已经存在一台Leader服务器）。
 *     针对这种情况，server1在试图选举Leader时，会被告知当前集群的Leader信息，此时server1只需要和Leader建立连接，并进行状态同步即可。
 *     具体流程：server1会给所有server推荐自己为Leader，所有server又会返回当前Leader的信息，如果server1收到的消息有半数以上都相同，则此时server1认为Leader已存在，server1只需要和Leader建立连接，并进行状态同步即可。
 *     代码详见：{@link FastLeaderElection#lookForLeader}方法中switch (n.state)的case LEADING分支。
 *  2、集群中确实不存在Leader。
 *
 *
 *  面试官：说说你对ZooKeeper集群与Leader选举的理解？  https://mp.weixin.qq.com/s/Z57YOfgoW60uff6013KbYg
 *
 */
public class FastLeaderElection implements Election {
    /**
     * leader选举从开始到结束 的超时时间
     * Determine how much time a process has to wait once it believes that it has reached the end of leader election.
     *
     * finalize：最终
     */
    final static int finalizeWait = 200;

    /**
     * 两次连续的通知检查之间的时间上限。
     * 这个时间决定系统在长时间的分片之后，系统会重新启动。
     * 默认为60秒。
     * Upper bound on the amount of time between two consecutive notification checks.
     * This impacts the amount of time to get the system up again after long partitions.
     * Currently 60 seconds.
     */
    final static int maxNotificationInterval = 60000;

    private static final Logger LOG = LoggerFactory.getLogger(FastLeaderElection.class);
    /**
     * 投票连接管理器。FastLeaderElection 基于TCP协议来连接两个节点(peers)之间的连接，并且就是由 QuorumCnxManager 来管理这样的连接。
     * Connection manager. Fast leader election uses TCP for communication between peers, and QuorumCnxManager manages such connections.
     */
    QuorumCnxManager manager;
    /**
     * 发送消息队列
     */
    LinkedBlockingQueue<ToSend> sendqueue;
    /**
     * 接收到消息的队列
     */
    LinkedBlockingQueue<Notification> recvqueue;
    QuorumPeer self;
    Messenger messenger;

    /**
     * 逻辑时钟
     * 在leader选举时，表示Epoch
     */
    AtomicLong logicalclock = new AtomicLong(); /* Election instance */

    /**
     * 提议哪个server id为Leader
     * （比如：想要推荐serverA节点为Leader）
     */
    long proposedLeader;
    /**
     * 协议事务id
     * （即：serverA节点最后的事务id）
     */
    long proposedZxid;
    /**
     * 提议为Leader的节点的epoch
     * （即：serverA节点epoch）
     */
    long proposedEpoch;

    /**
     * 是否停止选举
     */
    volatile boolean stop;

    /**
     * 构造方法。
     * 在每个ZooKeeper服务期间，FastLeaderElection类的对象，应该只被创建一次。
     * Such an object should be created only once by each peer during an instance of the ZooKeeper service.
     *
     * @param self  QuorumPeer that created this object
     * @param manager   Connection manager
     */
    public FastLeaderElection(QuorumPeer self, QuorumCnxManager manager) {
        this.stop = false;
        this.manager = manager;

        starter(self, manager);
    }

    /**
     * 这个方法只被构造方法调用。它是类对象构造的一部分。
     */
    private void starter(QuorumPeer self, QuorumCnxManager manager) {
        this.self = self;
        proposedLeader = -1;
        proposedZxid = -1;

        sendqueue = new LinkedBlockingQueue<ToSend>();
        recvqueue = new LinkedBlockingQueue<Notification>();
        this.messenger = new Messenger(manager);
    }

    static ByteBuffer buildMsg(int state, long leader, long zxid, long electionEpoch, long epoch) {
        byte requestBytes[] = new byte[40];
        ByteBuffer requestBuffer = ByteBuffer.wrap(requestBytes);

        requestBuffer.clear();
        requestBuffer.putInt(state);
        requestBuffer.putLong(leader);
        requestBuffer.putLong(zxid);
        requestBuffer.putLong(electionEpoch);
        requestBuffer.putLong(epoch);
        requestBuffer.putInt(Notification.CURRENTVERSION);

        return requestBuffer;
    }

    public long getLogicalClock() {
        return logicalclock.get();
    }

    /**
     * 清空收到消息队列
     *
     * @param v
     */
    private void leaveInstance(Vote v) {
        if (LOG.isDebugEnabled()) {
            LOG.debug("About to leave FLE instance: leader=" + v.getId() + ", zxid=0x" + Long.toHexString(v.getZxid()) + ", my id=" + self.getId() + ", my state=" + self.getPeerState());
        }
        recvqueue.clear();
    }

    public QuorumCnxManager getCnxManager() {
        return manager;
    }

    @Override
    public void shutdown() {
        stop = true;
        LOG.debug("Shutting down connection manager");
        manager.halt();
        LOG.debug("Shutting down messenger");
        messenger.halt();
        LOG.debug("FLE is down");
    }

    /**
     * 给所有的peer发送消息，用于修改我们的投票
     * Send notifications to all peers upon a change in our vote
     */
    private void sendNotifications() {
        for (QuorumServer server : self.getVotingView().values()) {
            long sid = server.id;
            ToSend notmsg = new ToSend(ToSend.mType.notification, proposedLeader, proposedZxid, logicalclock.get(), QuorumPeer.ServerState.LOOKING, sid, proposedEpoch);

            if (LOG.isDebugEnabled()) {
                LOG.debug("Sending Notification: " + proposedLeader + " (n.leader), 0x" + Long.toHexString(proposedZxid) + " (n.zxid), 0x" + Long.toHexString(logicalclock.get()) + " (n.round), " + sid + " (recipient), " + self.getId() + " (myid), 0x" + Long.toHexString(proposedEpoch) + " (n.peerEpoch)");
            }

            sendqueue.offer(notmsg);
        }
    }

    private void printNotification(Notification n) {
        LOG.info("Notification: " + n.toString() + self.getPeerState() + " (my state)");
    }

    /**
     * 选票PK
     * <pre>
     * 各个服务器都要投票进行 Leader 的选举，大家投票的标准是：
     *      epoch 大者投之； epoch 相等，则 zxid 大者投之；
     *      zxid 相等，则 myid 大者投之；最终，得票最多者当选。
     *      简单来说就是 zxid(epoch, zxid)大者投之，否则 myid 大者投之。
     * </pre>
     * @param newId service id
     * @param newZxid
     * @param newEpoch
     * @param curId
     * @param curZxid
     * @param curEpoch
     * @return
     */
    protected boolean totalOrderPredicate(long newId, long newZxid, long newEpoch, long curId, long curZxid, long curEpoch) {
        LOG.debug("id: " + newId + ", proposed id: " + curId + ", zxid: 0x" + Long.toHexString(newZxid) + ", proposed zxid: 0x" + Long.toHexString(curZxid));
        if (self.getQuorumVerifier().getWeight(newId) == 0) {
            return false;
        }
        /*
         * 返回true如果下面三个条件之一成立:
         * 1- New epoch is higher
         * 2- New epoch is the same as current epoch, but new zxid is higher
         * 3- New epoch is the same as current epoch, new zxid is the same as current zxid, but server id is higher.
         */
        return ((newEpoch > curEpoch) ||
                ((newEpoch == curEpoch) && ((newZxid > curZxid) ||
                        ((newZxid == curZxid) && (newId > curId)))));
    }

    /**
     * 是否终止选举
     * 给定一组选票votes，如果votes中有足够的选票和(第二个参数)vote相等，
     * 则表示从votes这些选票中，此时已经可以选举出一个leader，表示可以停止选举。
     * 此时返回true。
     * Termination predicate. Given a set of votes, determines if
     * have sufficient to declare the end of the election round.
     *
     * @param votes Set of votes。key is sid
     * @param vote
     * @return
     */
    protected boolean termPredicate(HashMap<Long, Vote> votes, Vote vote) {

        HashSet<Long> sidSet = new HashSet<Long>();

        for (Map.Entry<Long, Vote> entry : votes.entrySet()) {
            Vote v = entry.getValue();
            if (vote.equals(v)) {
                Long sid = entry.getKey();
                sidSet.add(sid);
            }
        }

        return self.getQuorumVerifier().containsQuorum(sidSet);
    }

    /**
     * 从一组投票中，判断第二个参数leader对应的sid，是否为Leader角色。
     * In the case there is a leader elected, and a quorum supporting
     * this leader, we have to check if the leader has voted and acked
     * that it is leading. We need this check to avoid that peers keep
     * electing over and over a peer that has crashed and it is no
     * longer leading.
     *
     * @param votes set of votes
     * @param   leader  leader id
     * @param   electionEpoch   epoch id
     */
    protected boolean checkLeader(HashMap<Long, Vote> votes, long leader, long electionEpoch) {

        // 断言
        boolean predicate = true;

        /*
         * If everyone else thinks I'm the leader, I must be the leader.
         * The other two checks are just for the case in which I'm not the
         * leader. If I'm not the leader and I haven't received a message
         * from leader stating that it is leading, then predicate is false.
         */
        if (leader != self.getId()) { // leader sid，不等于 当前server id
            if (votes.get(leader) == null) {
                predicate = false;
            } else if (votes.get(leader).getState() != ServerState.LEADING) {
                predicate = false;
            }
        } else if (logicalclock.get() != electionEpoch) {
            predicate = false;
        }

        return predicate;
    }

    /**
     * 这个方法从接收投票recv和待发出投票中，检查leader是否已经被选举出来。
     *
     * This predicate checks that a leader has been elected. It doesn't
     * make a lot of sense without context (check lookForLeader) and it
     * has been separated for testing purposes.
     *
     * @param recv  map of received votes
     * @param outofelection   map containing out of election votes (LEADING or FOLLOWING) key is sid
     * @param n     Notification
     * @return
     */
    protected boolean ooePredicate(HashMap<Long, Vote> recv, HashMap<Long, Vote> outofelection, Notification n) {
        // 收到消息的选票
        Vote notificationVote = new Vote(n.version, n.leader, n.zxid, n.electionEpoch, n.peerEpoch, n.state);

        return termPredicate(recv, notificationVote) && checkLeader(outofelection, n.leader, n.electionEpoch);
    }

    /**
     * 变更自己的选票
     * @param leader
     * @param zxid
     * @param epoch
     */
    synchronized void updateProposal(long leader, long zxid, long epoch) {
        if (LOG.isDebugEnabled()) {
            LOG.debug("Updating proposal: " + leader + " (newleader), 0x" + Long.toHexString(zxid) + " (newzxid), " + proposedLeader + " (oldleader), 0x" + Long.toHexString(proposedZxid) + " (oldzxid)");
        }
        proposedLeader = leader;
        proposedZxid = zxid;
        proposedEpoch = epoch;
    }

    synchronized Vote getVote() {
        return new Vote(proposedLeader, proposedZxid, proposedEpoch);
    }

    /**
     * A learning state can be either FOLLOWING or OBSERVING.
     * This method simply decides which one depending on the
     * role of the server.
     *
     * @return ServerState
     */
    private ServerState learningState() {
        if (self.getLearnerType() == LearnerType.PARTICIPANT) {
            LOG.debug("I'm a participant: " + self.getId());
            return ServerState.FOLLOWING;
        } else {
            LOG.debug("I'm an observer: " + self.getId());
            return ServerState.OBSERVING;
        }
    }

    /**
     * 返回表示server身份的投票初始值（myid）
     * Returns the initial vote value of server identifier.
     *
     * @return myid
     */
    private long getInitId() {
        if (self.getLearnerType() == LearnerType.PARTICIPANT) {
            return self.getId();
        } else {
            return Long.MIN_VALUE;
        }
    }

    /**
     * 返回记录到事务日志中，最大的事务id
     */
    private long getInitLastLoggedZxid() {
        if (self.getLearnerType() == LearnerType.PARTICIPANT) {
            return self.getLastLoggedZxid();
        } else {
            return Long.MIN_VALUE;
        }
    }

    /**
     * 返回当前选举server的选举epoch
     */
    private long getPeerEpoch() {
        if (self.getLearnerType() == LearnerType.PARTICIPANT) {
            try {
                return self.getCurrentEpoch();
            } catch (IOException e) {
                RuntimeException re = new RuntimeException(e.getMessage());
                re.setStackTrace(e.getStackTrace());
                throw re;
            }
        } else {
            return Long.MIN_VALUE;
        }
    }

    /**
     * 开始一个新的Leader选举周期。
     * 无论何时当前server的状态变为{@link ServerState#LOOKING}，这个方法就会被调用。
     * 这个方法会发送消息到集群中其他server节点。
     *
     * Starts a new round of leader election. Whenever our QuorumPeer
     * changes its state to LOOKING, this method is invoked, and it
     * sends notifications to all other peers.
     */
    @Override
    public Vote lookForLeader() throws InterruptedException {
        registJMX();

        try {
            // key是sid
            HashMap<Long, Vote> recvset = new HashMap<Long, Vote>();

            // Out Of Election。key是sid
            HashMap<Long, Vote> outofelection = new HashMap<Long, Vote>();

            // notTimeout 初始值为200，如果接受不到选举消息，其值就会不断翻倍。但最大为：maxNotificationInterval
            int notTimeout = finalizeWait;

            synchronized (this) {
                logicalclock.incrementAndGet();
                updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());
            }

            LOG.info("New election. My id =  " + self.getId() + ", proposed zxid=0x" + Long.toHexString(proposedZxid));

            // 给所有的peer发送投票消息
            sendNotifications();

            // 循环：交换选举投票信息，直到找到 leader
            while ((self.getPeerState() == ServerState.LOOKING) && (!stop)) {
                // 获取一个收到的消息
                Notification n = recvqueue.poll(notTimeout, TimeUnit.MILLISECONDS);

                /* ======================发送更多消息如果没有收到足够的消息，否则处理新投票消息。====================== */

                // 没有收到足够消息的情况：再次发送投票消息
                if (n == null) {
                    if (manager.haveDelivered()) {
                        sendNotifications();
                    } else {
                        manager.connectAll();
                    }

                    // 超时时间翻倍
                    int tmpTimeOut = notTimeout * 2;
                    notTimeout = (tmpTimeOut < maxNotificationInterval ? tmpTimeOut : maxNotificationInterval);
                    LOG.info("Notification time out: " + notTimeout);

                } else if (self.getVotingView().containsKey(n.sid)) {
                    // 收到投票消息：处理新投票消息。
                    switch (n.state) {
                        case LOOKING:
                            // 判断选举轮次：只有在同一选举轮次的投票才是有效的。
                            if (n.electionEpoch > logicalclock.get()) {
                                // 外部投票的选举轮次大于内部选举轮次时，replace logicalclock
                                logicalclock.set(n.electionEpoch); // replace：把[对方server的选举周期]设置为[当前server的选举周期]
                                recvset.clear(); // 清空已经收到的投票。（清空之前recvset中的选票都是以前选举轮次的。）

                                // 疑问:为什么 n.electionEpoch > logicalclock.get()时，还需要totalOrderPredicate()方法判断呢？不应该直接发送对方的投票消息么？毕竟人家已经开启了新一轮的leader选举。
                                //  解答：其实上面的 n.electionEpoch > logicalclock.get() 判断条件意思为：如果[发送消息server的选举周期n.electionEpoch]大于[当前server选举周期logicalclock]。
                                //       而不是 n.peerEpoch > proposedLeader。
                                //       【n.peerEpoch > proposedLeader这个判断条件表示：】
                                if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch, getInitId(), getInitLastLoggedZxid(), getPeerEpoch())) {
                                    updateProposal(n.leader, n.zxid, n.peerEpoch);
                                } else {
                                    updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());
                                }
                                sendNotifications();

                            } else if (n.electionEpoch < logicalclock.get()) {
                                if (LOG.isDebugEnabled()) {
                                    LOG.debug("Notification election epoch is smaller than logicalclock. n.electionEpoch = 0x" + Long.toHexString(n.electionEpoch) + ", logicalclock=0x" + Long.toHexString(logicalclock.get()));
                                }
                                break;

                            } else if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch, proposedLeader, proposedZxid, proposedEpoch)) {
                                // 收到消息逻辑时钟和当前server逻辑时钟相同：再使用 totalOrderPredicate() 方法判断对方投票是不是更高，
                                // 如果是，则修改自己的投票，并发送出去。
                                updateProposal(n.leader, n.zxid, n.peerEpoch);
                                sendNotifications();
                            }

                            if (LOG.isDebugEnabled()) {
                                LOG.debug("Adding vote: from=" + n.sid + ", proposed leader=" + n.leader + ", proposed zxid=0x" + Long.toHexString(n.zxid) + ", proposed election epoch=0x" + Long.toHexString(n.electionEpoch));
                            }

                            // 消息归档：归档接收到的消息。
                            recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch));

                            // 统计投票：为了统计集群中是否已经有过半的server认可了当前的内部投票。如果已经有过半的server认可该内部投票，则终止投票，表示此时可选出Leader。
                            if (termPredicate(recvset, new Vote(proposedLeader, proposedZxid, logicalclock.get(), proposedEpoch))) {

                                // 从 recvqueue 队列中获取一个消息。（超时时间200毫秒）
                                //  1、如果外部选票 K.O 当前内部选票：则把消息还放进 recvqueue 队列中，然后break。（说明recvqueue队列中还有更牛逼的选票，Leader选举继续）
                                //  2、如果recvqueue队列中所有选票都不比内部选票牛逼,则从recvqueue中获取消息超时后(超时事件200毫秒)，下面if (n == null)条件就会成立。
                                //     此时内部选票对应的proposedLeader就是Leader服务器的sid。
                                while ((n = recvqueue.poll(finalizeWait, TimeUnit.MILLISECONDS)) != null) {
                                    if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch, proposedLeader, proposedZxid, proposedEpoch)) {
                                        recvqueue.put(n);
                                        break;
                                    }
                                }

                                // 根据上面注释，这个条件成立则说明：当前内部选票可以选出Leader
                                if (n == null) {
                                    // 确定当前server状态：如果 proposedLeader 就是自己sid，则当前server就是LEADING，否则是FOLLOWING或OBSERVING
                                    self.setPeerState((proposedLeader == self.getId()) ? ServerState.LEADING : learningState());

                                    // Leader对应的选票
                                    Vote endVote = new Vote(proposedLeader, proposedZxid, logicalclock.get(), proposedEpoch);
                                    leaveInstance(endVote); // Leader已经选出来了：情况收到的所有选票
                                    return endVote;
                                    // 这里有个细节：如果统计投票发现已经有过半的server认可了当前选票，这是zk并不会立即更改当前server状态。而是等一会（默认200毫秒）来确定是否有新的更优的投票。
                                }
                            }
                            break;
                        case OBSERVING:
                            LOG.debug("Notification from observer: " + n.sid);
                            break;
                        case FOLLOWING:
                        case LEADING:
                            // 发送消息的server已经选出Leader节点了。详细注释见此类注释。

                            // 当前server选举时钟和收到消息的时钟相同情况：
                            if (n.electionEpoch == logicalclock.get()) {
                                // 把收到的消息n，放到recvset中。
                                recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch));

                                // 判断当前收到投票和待发投票中，能否决定Leader已经选出。
                                if (ooePredicate(recvset, outofelection, n)) {
                                    // 设置当前server状态。
                                    self.setPeerState((n.leader == self.getId()) ? ServerState.LEADING : learningState());

                                    // 经过Leader选举过程，选出的Leader对应的信息。
                                    Vote endVote = new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch);
                                    // Leader已经选出来了，清空接收消息队列
                                    leaveInstance(endVote);
                                    return endVote; // 返回选出Leader服务器的信息。
                                }
                            }

                            // 当前server逻辑时钟和Leader的逻辑时钟不一样：
                            Vote vote = new Vote(n.version, n.leader, n.zxid, n.electionEpoch, n.peerEpoch, n.state);
                            outofelection.put(n.sid, vote);

                            // 通过选票数量，判断Leader是否能被确定。
                            // （就算收到Leader服务器发来的消息说它自己就是Leader，也要保证超过半数以上的Follower认同此Leader）
                            if (ooePredicate(outofelection, outofelection, n)) {
                                synchronized (this) {
                                    // 把当前server的逻辑时钟，设置为Leader的逻辑时钟
                                    logicalclock.set(n.electionEpoch);
                                    // 设置当前server状态
                                    self.setPeerState((n.leader == self.getId()) ? ServerState.LEADING : learningState());
                                }
                                Vote endVote = new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch);
                                leaveInstance(endVote);
                                return endVote;
                            }
                            break;
                        default:
                            LOG.warn("Notification state unrecognized: {} (n.state), {} (n.sid)", n.state, n.sid);
                            break;
                    }
                } else {
                    LOG.warn("Ignoring notification from non-cluster member " + n.sid);
                }
            }
            return null;
        } finally {
            try {
                if (self.jmxLeaderElectionBean != null) {
                    MBeanRegistry.getInstance().unregister(self.jmxLeaderElectionBean);
                }
            } catch (Exception e) {
                LOG.warn("Failed to unregister with JMX", e);
            }
            self.jmxLeaderElectionBean = null;
            LOG.debug("Number of connection processing threads: {}", manager.getConnectionThreadCount());
        }
    }

    private void registJMX() {
        try {
            self.jmxLeaderElectionBean = new LeaderElectionBean();
            MBeanRegistry.getInstance().register(self.jmxLeaderElectionBean, self.jmxLocalPeerBean);
        } catch (Exception e) {
            LOG.warn("Failed to register with JMX", e);
            self.jmxLeaderElectionBean = null;
        }
        if (self.start_fle == 0) {
            // 设置Leader选举开始时间
            self.start_fle = Time.currentElapsedTime();
        }
    }

    /**
     * <pre>
     * Notifications are messages that let other peers know that
     * a given peer has changed its vote, either because it has
     * joined leader election or because it learned of another
     * peer with higher zxid or same zxid and higher server id
     *
     * 在选举时发送的消息。
     *
     * 举例：
     * 如果serverA/serverB/serverC三个节点组成一个集群（quorum），
     * 经过一段时间的投票（vote），假设serverA和serverB两个节点投票状态变为：
     *  1、serverA给serverB发消息n，提议serverC为Leader；
     *  2、serverB提议serverA为Leader；serverB收到了serverA的消息n。
     * 则此时 serverB 的状态：
     *     proposedLeader = sid(serverA)         // 由于serverB提议serverA为Leader，则proposedLeader 为 serverA 的sid。
     *     proposedZxid = zxid(serverA)          // 由于serverB提议serverA为Leader，则proposedZxid 为 serverA 最后事务id。
     *     proposedEpoch = epoch(serverA)        // 由于serverB提议serverA为Leader，则proposedEpoch 为 serverA 的epoch。
     *     n.sid = sid(serverA);                 // 由于消息n是serverA发送的，则n.sid为 serverA 的sid。
     *     n.electionEpoch = eopch(serverA);     // 由于消息n是serverA发送的，则n.electionEpoch就为 serverA 的epoch。
     *     n.leader = sid(serverC);              // 由于serverA提议serverC为Leader，则n.leader为 serverC 的sid。
     *     n.peerEpoch = eopch(serverC);         // 由于serverA提议serverC为Leader，则n.peerEpoch为 serverC 的epoch。
     *     n.zxid = zxid(serverC);               // 由于serverA提议serverC为Leader，则n.zxid为 serverC 最后事务id。
     * </pre>
     */
    static public class Notification {
        /**
         * Format version, introduced in 3.4.6
         */
        public final static int CURRENTVERSION = 0x1;

        /**
         * message format version
         */
        int version;

        /**
         * 被推举的Leader的sid值
         * Proposed leader
         *
         * @see Vote#id
         */
        long leader;

        /**
         * 被推举的Leader的事务id
         * zxid of the proposed leader
         *
         * @see Vote#zxid
         */
        long zxid;

        /**
         * 选举轮次：只有在同一选举轮次的投票才是有效的。
         * @see Vote#electionEpoch
         */
        long electionEpoch;

        /**
         * 当前发送消息的服务器的状态
         * @see Vote#state
         */
        QuorumPeer.ServerState state;

        /**
         * 发送消息的server的sid
         * Address of sender
         */
        long sid;

        /**
         * 被选举的Leader的epoch
         * epoch of the proposed leader
         *
         * @see Vote#peerEpoch
         */
        long peerEpoch;

        @Override
        public String toString() {
            return Long.toHexString(version) + " (message format version), " + leader + " (n.leader), 0x" + Long.toHexString(zxid) + " (n.zxid), 0x" + Long.toHexString(electionEpoch) + " (n.round), " + state + " (n.state), " + sid + " (n.sid), 0x" + Long.toHexString(peerEpoch) + " (n.peerEpoch) ";
        }
    }

    /**
     * Messages that a peer wants to send to other peers.
     * These messages can be both Notifications and Acks
     * of reception of notification.
     */
    static public class ToSend {
        /*
         * Proposed leader in the case of notification
         */
        long leader;
        /*
         * id contains the tag for acks, and zxid for notifications
         */
        long zxid;
        /*
         * Epoch
         */
        long electionEpoch;
        /*
         * Current state;
         */
        QuorumPeer.ServerState state;
        /*
         * Address of recipient
         */
        long sid;
        /*
         * Leader epoch
         */
        long peerEpoch;

        ToSend(mType type, long leader, long zxid, long electionEpoch, ServerState state, long sid, long peerEpoch) {

            this.leader = leader;
            this.zxid = zxid;
            this.electionEpoch = electionEpoch;
            this.state = state;
            this.sid = sid;
            this.peerEpoch = peerEpoch;
        }

        /**
         * 消息类别
         */
        enum mType {crequest, challenge, notification, ack}
    }

    /**
     * Multi-threaded implementation of message handler. Messenger
     * implements two sub-classes: WorkReceiver and  WorkSender. The
     * functionality of each is obvious from the name. Each of these
     * spawns a new thread.
     */

    protected class Messenger {

        WorkerSender ws;
        WorkerReceiver wr;

        /**
         * Constructor of class Messenger.
         *
         * @param manager   Connection manager
         */
        Messenger(QuorumCnxManager manager) {

            this.ws = new WorkerSender(manager);

            Thread t = new Thread(this.ws, "WorkerSender[myid=" + self.getId() + "]");
            t.setDaemon(true);
            t.start();

            this.wr = new WorkerReceiver(manager);

            t = new Thread(this.wr, "WorkerReceiver[myid=" + self.getId() + "]");
            t.setDaemon(true);
            t.start();
        }

        /**
         * Test if both send and receive queues are empty.
         */
        public boolean queueEmpty() {
            return (sendqueue.isEmpty() || recvqueue.isEmpty());
        }

        /**
         * Stops instances of WorkerSender and WorkerReceiver
         */
        void halt() {
            this.ws.stop = true;
            this.wr.stop = true;
        }

        /**
         * Receives messages from instance of QuorumCnxManager on
         * method run(), and processes such messages.
         */

        class WorkerReceiver extends ZooKeeperThread {
            volatile boolean stop;
            QuorumCnxManager manager;

            WorkerReceiver(QuorumCnxManager manager) {
                super("WorkerReceiver");
                this.stop = false;
                this.manager = manager;
            }

            @Override
            public void run() {

                Message response;
                while (!stop) {
                    // Sleeps on receive
                    try {
                        response = manager.pollRecvQueue(3000, TimeUnit.MILLISECONDS);
                        if (response == null) {
                            continue;
                        }

                        /*
                         * If it is from an observer, respond right away.
                         * Note that the following predicate assumes that
                         * if a server is not a follower, then it must be
                         * an observer. If we ever have any other type of
                         * learner in the future, we'll have to change the
                         * way we check for observers.
                         */
                        if (!self.getVotingView().containsKey(response.sid)) {
                            Vote current = self.getCurrentVote();
                            ToSend notmsg = new ToSend(ToSend.mType.notification, current.getId(), current.getZxid(), logicalclock.get(), self.getPeerState(), response.sid, current.getPeerEpoch());

                            sendqueue.offer(notmsg);
                        } else {
                            // Receive new message
                            if (LOG.isDebugEnabled()) {
                                LOG.debug("Receive new notification message. My id = " + self.getId());
                            }

                            /*
                             * We check for 28 bytes for backward compatibility
                             */
                            if (response.buffer.capacity() < 28) {
                                LOG.error("Got a short response: " + response.buffer.capacity());
                                continue;
                            }
                            boolean backCompatibility = (response.buffer.capacity() == 28);
                            response.buffer.clear();

                            // Instantiate Notification and set its attributes
                            Notification n = new Notification();

                            // State of peer that sent this message
                            QuorumPeer.ServerState ackstate = QuorumPeer.ServerState.LOOKING;
                            switch (response.buffer.getInt()) {
                                case 0:
                                    ackstate = QuorumPeer.ServerState.LOOKING;
                                    break;
                                case 1:
                                    ackstate = QuorumPeer.ServerState.FOLLOWING;
                                    break;
                                case 2:
                                    ackstate = QuorumPeer.ServerState.LEADING;
                                    break;
                                case 3:
                                    ackstate = QuorumPeer.ServerState.OBSERVING;
                                    break;
                                default:
                                    continue;
                            }

                            n.leader = response.buffer.getLong();
                            n.zxid = response.buffer.getLong();
                            n.electionEpoch = response.buffer.getLong();
                            n.state = ackstate;
                            n.sid = response.sid;
                            if (!backCompatibility) {
                                n.peerEpoch = response.buffer.getLong();
                            } else {
                                if (LOG.isInfoEnabled()) {
                                    LOG.info("Backward compatibility mode, server id=" + n.sid);
                                }
                                n.peerEpoch = ZxidUtils.getEpochFromZxid(n.zxid);
                            }

                            /*
                             * Version added in 3.4.6
                             */

                            n.version = (response.buffer.remaining() >= 4) ?
                                    response.buffer.getInt() : 0x0;

                            /*
                             * Print notification info
                             */
                            if (LOG.isInfoEnabled()) {
                                printNotification(n);
                            }

                            /*
                             * If this server is looking, then send proposed leader
                             */

                            if (self.getPeerState() == QuorumPeer.ServerState.LOOKING) {
                                recvqueue.offer(n);

                                /*
                                 * Send a notification back if the peer that sent this
                                 * message is also looking and its logical clock is
                                 * lagging behind.
                                 */
                                if ((ackstate == QuorumPeer.ServerState.LOOKING)
                                        && (n.electionEpoch < logicalclock.get())) {
                                    Vote v = getVote();
                                    ToSend notmsg = new ToSend(ToSend.mType.notification, v.getId(), v.getZxid(), logicalclock.get(), self.getPeerState(), response.sid, v.getPeerEpoch());
                                    sendqueue.offer(notmsg);
                                }
                            } else {
                                /*
                                 * If this server is not looking, but the one that sent the ack
                                 * is looking, then send back what it believes to be the leader.
                                 */
                                Vote current = self.getCurrentVote();
                                if (ackstate == QuorumPeer.ServerState.LOOKING) {
                                    if (LOG.isDebugEnabled()) {
                                        LOG.debug("Sending new notification. My id =  " + self.getId() + " recipient=" + response.sid + " zxid=0x" + Long.toHexString(current.getZxid()) + " leader=" + current.getId());
                                    }

                                    ToSend notmsg;
                                    if (n.version > 0x0) {
                                        notmsg = new ToSend(
                                                ToSend.mType.notification, current.getId(), current.getZxid(), current.getElectionEpoch(), self.getPeerState(), response.sid, current.getPeerEpoch());

                                    } else {
                                        Vote bcVote = self.getBCVote();
                                        notmsg = new ToSend(
                                                ToSend.mType.notification, bcVote.getId(), bcVote.getZxid(), bcVote.getElectionEpoch(), self.getPeerState(), response.sid, bcVote.getPeerEpoch());
                                    }
                                    sendqueue.offer(notmsg);
                                }
                            }
                        }
                    } catch (InterruptedException e) {
                        System.out.println("Interrupted Exception while waiting for new message" + e.toString());
                    }
                }
                LOG.info("WorkerReceiver is down");
            }
        }

        /**
         * This worker simply dequeues a message to send and
         * and queues it on the manager's queue.
         */

        class WorkerSender extends ZooKeeperThread {
            volatile boolean stop;
            QuorumCnxManager manager;

            WorkerSender(QuorumCnxManager manager) {
                super("WorkerSender");
                this.stop = false;
                this.manager = manager;
            }

            @Override
            public void run() {
                while (!stop) {
                    try {
                        ToSend m = sendqueue.poll(3000, TimeUnit.MILLISECONDS);
                        if (m == null) {
                            continue;
                        }

                        process(m);
                    } catch (InterruptedException e) {
                        break;
                    }
                }
                LOG.info("WorkerSender is down");
            }

            /**
             * Called by run() once there is a new message to send.
             *
             * @param m     message to send
             */
            void process(ToSend m) {
                ByteBuffer requestBuffer = buildMsg(m.state.ordinal(), m.leader, m.zxid, m.electionEpoch, m.peerEpoch);
                manager.toSend(m.sid, requestBuffer);
            }
        }

    }
}
